package vip.zhenzicheng.rocketmq.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 生产者
 *
 * @author zhenzicheng
 * @date 2022-05-28 19:34
 */
@Component
@Slf4j
public class RocketMQProducer {

  @Resource
  private RocketMQTemplate rocketMQTemplate;

  /**
   * 简单消息
   */
  public void sendMessage(String topic, String msg) {
    org.apache.rocketmq.common.message.Message message = new org.apache.rocketmq.common.message.Message();
    DefaultMQProducer producer = new DefaultMQProducer();
    message.setBody(msg.getBytes());
    // 消息延迟200ms后发送
    message.setDelayTimeLevel(200);
    message.setTopic(topic);
    // 等待消息存储成功
    message.setWaitStoreMsgOK(true);
    // TAG
    message.setTags("SEARCH");
    message.setKeys("user");
    try {
      producer.send(message, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
          log.info("发送成功");
        }

        @Override
        public void onException(Throwable throwable) {
          log.error("出现异常");
        }
      });
    } catch (MQClientException | RemotingException | InterruptedException e) {
      e.printStackTrace();
    }
  }

  /**
   * 事务消息
   */
  public void sendMessageInTransaction(String topic, String msg) throws InterruptedException {
    String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
    for (int i = 0; i < 10; i++) {
      // 尝试在Header中加入一些自定义的属性。
      Message<String> message = MessageBuilder.withPayload(msg)
        .setHeader(RocketMQHeaders.TRANSACTION_ID, "TransID_" + i)
        // 发到事务监听器里后，这个自己设定的TAGS属性会丢失。但是上面那个属性不会丢失。
        .setHeader(RocketMQHeaders.TAGS, tags[i % tags.length])
        // MyProp在事务监听器里也能拿到，为什么就单单这个RocketMQHeaders.TAGS拿不到？这只能去调源码了。
        .setHeader("MyProp", "MyProp_" + i)
        .build();
      // 与Spring整合以后TAG属性嵌入进destination，格式为 Topic:Tag
      String destination = topic + ":" + tags[i % tags.length];
      // 发送事务消息时，还是会转换成RocketMQ的Message对象
      SendResult sendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, destination);
      System.out.printf("%s%n", sendResult);

      Thread.sleep(10);
    }
  }

  /**
   * 对象消息，同步发送
   *
   * @param destination 格式为: Topic:TAG
   * @return String返回值
   */
  public String sendAndReceive(String destination, Object obj) {
    Message<Object> message = MessageBuilder.withPayload(obj).build();
    return rocketMQTemplate.sendAndReceive(destination, message, String.class);
  }
}
